/*
 * LamportMutex.cpp
 *
 *  Created on: Nov 23, 2012
 *      Author: dan
 */

#include "LamportMutex.h"
#include "Print.hpp"

namespace {

typedef boost::asio::ip::tcp::socket socket_t;
typedef boost::shared_ptr<socket_t> socket_ref;
typedef std::map<std::string, socket_ref> Hosts;
typedef boost::system::error_code error_code;
typedef boost::thread thread_t;
typedef boost::shared_ptr<thread_t> thread_ref;
typedef std::map<std::string, thread_ref> Threads;
typedef boost::asio::io_service ios_t;
typedef boost::shared_ptr<ios_t> ios_ref;

}

LamportMutex::LamportMutex(Network& network, Network::Team team, std::string m_name) :
		network(network), team(team), name(m_name), time(0), state(MTX_OUT) {
	boost::mutex::scoped_lock l(mtx);
	callbackId = network.setCallbacks(
			boost::bind(&LamportMutex::Event_OnConnection, this, _1, _2),
			boost::bind(&LamportMutex::Event_OnDisconnection, this, _1, _2),
			boost::bind(&LamportMutex::Event_OnMessage, this, _1, _2, _3));
	init();
}
LamportMutex::~LamportMutex() {
	network.removeCallbacks(callbackId);
}

void LamportMutex::Event_OnConnection(std::string id, error_code error) {
	PRINT("LamportMutex("+name+"): EVENT<CONNECTION>: id="<<id<<", error="<<error);
	if(error)return;
	boost::mutex::scoped_lock l(mtx);

	if (inTeam(id) == false)
		return;
	connected[id] = true;
	PRINT("notify about changes.");
	state_changed.notify_one();
}
void LamportMutex::Event_OnDisconnection(std::string id, error_code error) {
	PRINT("LamportMutex("+name+"): EVENT<DISCONNECTION>: id="<<id<<", error="<<error);
	//if(error) return;
	boost::mutex::scoped_lock l(mtx);

	if (inTeam(id) == false)
		return;
	connected[id] = false;
	ready[id] = false;
	PRINT("notify about changes.");
	state_changed.notify_one();
}
void LamportMutex::Event_OnMessage(std::string id, std::string message, error_code error) {
	PRINT("LamportMutex("+name+"): EVENT<MESSAGE>: id="<<id<<", error="<<error<<", message="<<message);
	if(error) return;

	boost::mutex::scoped_lock l(mtx);

	if (inTeam(id) == false || isMutexMessage(message)==false){
		PRINT("this message not from my team or it's not a mutex message");
		return;
	}
	size_t mtime = getTime(message);
	time = (mtime>time?mtime:time)+1;
	PRINT("Current time = "<<time);
	PRINT("Message type is "<<getType(message));
	std::string messageType = getType(message);
	if(messageType==std::string("ok")){
		PRINT("OK from "<<id);
		ready[id] = true;
	}else if(messageType==std::string("req")){
		PRINT("REQUEST from "<<id<<" with time="<<mtime);
		error_code send_error;
		if(state==MTX_IN){
			PRINT("I am in section. add to queue");
			waitingList.push_back(id);
		}else if(state==MTX_OUT){
			PRINT("I am not in section. send OK");
			network.send(id, getMutexMessage("ok",time), send_error);
		}else if(state==MTX_WAIT){
			if(mtime<sentTime || (mtime==sentTime && network.MeLessThen(id))){
				PRINT("I am wait for section, but my request time "<<sentTime<<". send OK");
				network.send(id, getMutexMessage("ok",time), send_error);
			}else{
				PRINT("I am wait for section, but my request time "<<sentTime<<". add to queue");
				waitingList.push_back(id);
			}
		}
	}else{
		PRINT("unknown message type");
	}
	PRINT("notify about changes.");
	state_changed.notify_one();

}

void LamportMutex::waitConnection() {
	boost::mutex::scoped_lock l(mtx);
	while (true) {
		if (allConnected() == true)
			return;
		state_changed.wait(l);
	}
}
void LamportMutex::waitConnection(boost::mutex::scoped_lock& ext_l) {
	boost::mutex::scoped_lock l(mtx);
	bool ext_l_locked = true;
	while (true) {
		if (allConnected() == true) {
			if (ext_l_locked == false)
				ext_l.lock();
			return;
		}
		ext_l_locked = false;
		ext_l.unlock();
		state_changed.wait(l);
	}
}

bool LamportMutex::lock() {
	boost::mutex::scoped_lock l(mtx);
	state=MTX_WAIT;
	network.send(team, getMutexMessage("req",time));
	sentTime = time;
	while (true) {
		if (allConnected() == false) {
			PRINT("LamportMutex("+name+"): allConnected = false");
			init();
			state = MTX_OUT;
			return false;
		}
		if (allReady()) {
			PRINT("LamportMutex("+name+"): allReady() = true");
			init();
			state = MTX_IN;
			return true;
		}
		PRINT("LamportMutex("+name+"): wait for changes.");
		state_changed.wait(l);
		PRINT("LamportMutex("+name+"): something changed in mutex status");
	}
	init();
	state = MTX_OUT;
	return false;
}


void LamportMutex::unlock(){
	boost::mutex::scoped_lock l(mtx);
	network.send(waitingList, getMutexMessage("ok",time));
	waitingList.clear();
	state = MTX_OUT;
}

bool LamportMutex::allConnected() {
	for (HostList::iterator i = connected.begin(); i != connected.end(); i++)
		if (i->second == false)
			return false;
	return true;
}
bool LamportMutex::allReady() {
	for (HostList::iterator i = ready.begin(); i != ready.end(); i++)
		if (i->second == false)
			return false;
	return true;
}
void LamportMutex::init() {
	ready.clear();
	teamSet.clear();
	std::set<std::string> active = network.getConnectionsSet();
	for (size_t i = 0; i < team.size(); i++) {
		std::string member = team[i].str();
		connected[member] = active.find(member) != active.end()
				|| network.isMe(member);
		ready[member] = false || network.isMe(member);
		teamSet.insert(member);
	}
	printTeamStatus();
}
void LamportMutex::printTeamStatus() {
	for (size_t i = 0; i < team.size(); i++) {
		std::string member = team[i].str();
		PRINT(
				"LamportMutex("+name+"): team member ["+member+"] is " <<(connected[member]?"connected":"disconnected")<<" and " <<(ready[member]?"ready":"not ready"));
	}
}
bool LamportMutex::inTeam(std::string id) {
	return teamSet.find(id) != teamSet.end();
}

std::string LamportMutex::getMutexMessage(std::string type,size_t time)const{
	std::stringstream s; s<<"@mutex:"<<name<<":"<<type<<":"<<time;
	return s.str();
}
bool LamportMutex::isMutexMessage(const std::string& message)const{
	size_t col = message.find_last_of(':');
	if(col==std::string::npos || col==message.size()-1) return false;
	std::string mespref = message.substr(0,col);
	col = mespref.find_last_of(':');
	if(col==std::string::npos || col==mespref.size()-1) return false;
	std::string mesprefpref = mespref.substr(0,col);
	return mesprefpref==("@mutex:"+name);
}
size_t LamportMutex::getTime(const std::string& message)const{
	size_t col = message.find_last_of(':');
	if(col==std::string::npos || col==message.size()-1) return 0;
	std::stringstream s; s<<message.substr(col+1);
	size_t n; s>>n;
	return n;
}
std::string LamportMutex::getType(const std::string& message)const{
	size_t col = message.find_last_of(':');
	if(col==std::string::npos || col==message.size()-1) return 0;
	std::string p = message.substr(0,col);
	col = p.find_last_of(':');
	if(col==std::string::npos || col==message.size()-1) return 0;
	return p.substr(col+1);
}










